package cn.doitedu.cn.doitedu.rtdw.data_sync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_GeoHashTable2Hbase {
    public static void main(String[] args) {


        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 1.建表，映射mysql中的ref_geohash表
        tenv.executeSql(
                " create table ref_geohash_mysql(     " +
                        "    geohash string,                  " +
                        "    province string,                 " +
                        "    city string,                     " +
                        "    region string,                   " +
                        "    PRIMARY KEY (geohash) NOT ENFORCED    " +
                        " ) with (                                 " +
                        "  'connector' = 'jdbc',                            "+
                        "  'url' = 'jdbc:mysql://doitedu:3306/realtimedw',  "+
                        "  'table-name' = 'ref_geohash',                    "+
                        "  'username' = 'root',                             "+
                        "  'password' = 'root'                              "+
                        " )                                                 "
        );

        // 2.建表，映射hbase中的目标表
        tenv.executeSql(
                " create table dim_geohash_hbase(              " +
                        "   geohash string,                       " +
                        " 	f row<                                " +
                        " 	province string,                      " +
                        " 	city string,                          " +
                        " 	region string>,                       " +
                        " 	PRIMARY KEY (geohash) NOT ENFORCED    " +
                        " ) with (                                 " +
                        "  'connector' = 'hbase-2.2',              " +
                        "  'table-name' = 'dim_ref_geohash',        " +
                        "  'zookeeper.quorum' = 'doitedu:2181'     " +
                        " )                                        "
        );


        // 3.写sql
        tenv.executeSql("insert into dim_geohash_hbase select geohash,row(province,city,region) from  ref_geohash_mysql");


    }


}


